package tcpsubpub

import (
	"fmt"
	"gitee.com/ymofen/gobase"
	"gitee.com/ymofen/gobase/factory"
	"gitee.com/ymofen/gobase/gotask"
	"gitee.com/ymofen/gobase/subpub"
	"io"
	"sync/atomic"
)

// 将数据分流给NtripClient
type PubTcpServerSession struct {
	closedflag int32
	notify     subpub.ISubPub
	sessionid  string
	mp         string

	conf gobase.StrMap

	copybufflag int8

	operaRef   *gobase.ObjectInnerRef
	opera      *TcpServerNode
	pubmax     int
	slave      int8
	mpflag     int8
	user, pass string
}

func CreatePubTcpServerSession(args ...interface{}) (rval interface{}, err error) {
	plg := &PubTcpServerSession{
		sessionid: args[0].(string),
		conf:      args[1].(gobase.StrMap),
		notify:    args[2].(subpub.ISubPub),
	}
	return plg, nil
}

func (this *PubTcpServerSession) Pub(topic string, max int, args ...interface{}) int {
	if len(args) == 0 {
		this.notify.Pub(this.sessionid, 0, subpub.ACTION_ERROR, fmt.Errorf("没有要Pub的数据"))
		return 0
	}

	opera := this.opera
	if opera == nil {
		return 0
	}

	if buf, ok := args[0].([]byte); ok {
		return opera.PubBuf(this.pubmax, buf)
	} else if str, ok := args[0].(string); ok {
		return opera.PubBuf(this.pubmax, []byte(str))
	} else {
		this.notify.Pub(this.sessionid, 0, subpub.ACTION_ERROR, fmt.Errorf("参数不是[]byte类型"))
		return 0
	}
}

func (this *PubTcpServerSession) PostData(args ...interface{}) error {
	return nil
}

func (this *PubTcpServerSession) onPubEvt(id, topic string, args ...interface{}) (ok bool) {
	ex := this.notify
	if ex == nil {
		return false
	}

	if len(args) < 2 {
		return false
	}
	action := args[0].(int)
	if action == subpub.ACTION_SERVER_CONN_RECV_BUF {
		return false
	}

	ex.Pub(this.sessionid, 0, args...)
	return true
}

// 连接参数 network=tcp4&maxonline=-1&connstr=:15005
//

func (this *PubTcpServerSession) Start() {

	s0 := this.conf.StringByName("connstr", "")
	connf := gobase.ParseConnString(s0)
	network := this.conf.StringByName("network", "")
	if len(network) == 0 {
		network = connf.Type
	}
	if len(network) == 0 {
		network = "tcp4"
	}

	connstr := fmt.Sprintf("network=%s&addr=%s",
		network, connf.Addr)

	this.operaRef = DefaultTcpServerGroup.CheckGetRef(connstr)
	this.opera = this.operaRef.GetObject().(*TcpServerNode)
	this.opera.ex.Sub(this.sessionid, "evt", this.onPubEvt)

	maxonline := this.conf.IntByName("maxonline", -1)
	if maxonline != -1 {
		this.opera.maxonline = maxonline
	}

	this.notify.Pub(this.sessionid, 0, subpub.ACTION_CONNECTED)
	this.opera.checkStart()
}

func (this *PubTcpServerSession) Close() error {
	if !atomic.CompareAndSwapInt32(&this.closedflag, 0, 1) {
		return io.ErrClosedPipe
	}
	if this.operaRef != nil {
		this.opera.ex.Unsub(this.sessionid, "evt")
		this.opera = nil

		DefaultTcpServerGroup.ReleaseRef(this.operaRef)
		this.operaRef = nil
	}

	gotask.DefaultPatrolTask().DelTask(this.sessionid)

	this.notify.Pub(this.sessionid, 0, subpub.ACTION_DISCONNECTED)
	this.notify = nil
	return nil
}

func init() {
	factory.RegisterCreateFunc("pub.tcpserver", CreatePubTcpServerSession)
}
